package com.ilxqx.framework.amqp.service.impl;

import com.ilxqx.framework.amqp.constant.RabbitMessageConstants;
import com.ilxqx.framework.amqp.entity.SysMessageLog;
import com.ilxqx.framework.amqp.enumeration.SysMessageLogStatusEnum;
import com.ilxqx.framework.amqp.po.EmailSendingTask;
import com.ilxqx.framework.amqp.po.SysRabbitMessage;
import com.ilxqx.framework.amqp.service.MessageSenderService;
import com.ilxqx.framework.amqp.service.SysMessageLogService;
import com.ilxqx.framework.exception.FrameworkException;
import com.ilxqx.framework.util.ObjectUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Optional;

import static org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;

/**
 * 系统任务消息发送服务类
 * @author venus
 */
@Service
@Slf4j
public class MessageSenderServiceImpl implements MessageSenderService {

    @Value("${spring.rabbitmq.template.exchange}")
    private String defaultExchange;

    @Value("${spring.rabbitmq.template.routing-key}")
    private String defaultRoutingKey;

    /**
     * RabbitMQ操作模板类
     */
    private final RabbitTemplate template;

    private final SysMessageLogService sysMessageLogService;

    private final ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
        if (correlationData == null) {
            // 处理失败
            return;
        }
        String messageId = correlationData.getId();
        if (messageId == null) {
            // 没有消息ID，同样处理失败
            return;
        }
        log.info("RabbitMQ收到系统投递的消息，并返回确认消息[{}]", correlationData.getId());

        if (ack) {
            // 返回成功，更新状态
            this.handleSuccess(Long.parseLong(messageId));
            return;
        }
        // 失败
        log.warn("消息投递失败[{}]", cause);
        this.handleFailure(Long.parseLong(messageId));
    };

    private final ReturnCallback returnCallback = (message, replyCode, replyText, exchange, routingKey) -> {
        Long messageId = Long.valueOf(message.getMessageProperties().getHeaders().get("spring_returned_message_correlation").toString());
        log.warn("Rabbit返回结果，消息[{}]处理失败，失败码[{}]失败信息[{}]，交换机[{}]路由key[{}]", messageId, replyCode, replyText, exchange, routingKey);
        // 设置处理失败
        this.returnFailure(messageId);
    };

    private void check(Long messageId) {
        Optional<SysMessageLog> sysMessageLogOptional = this.sysMessageLogService.get(messageId);
        if (sysMessageLogOptional.isPresent()) {
            SysMessageLog sysMessageLog = sysMessageLogOptional.get();
            if (!sysMessageLog.getStatus().equals(SysMessageLogStatusEnum.SENDING_SUCCESS) && sysMessageLog.getTryCount() == RabbitMessageConstants.MAX_RETRY_COUNT) {
                // 说明已经失败了
                EmailSendingTask task = new EmailSendingTask("2929923469@qq.com", "有RabbitMQ消息失败了", "消息标题：" + sysMessageLog.getTitle() + ", 消息发起者：" + sysMessageLog.getOperationUserName() + ", 消息状态：" + sysMessageLog.getStatus().getMeaning() + ", 创建时间：" + sysMessageLog.getCreateTime());
                sendMessage("消息失败邮件通知", task);
            }
        }
    }

    private void handleFailure(Long messageId) {
        check(messageId);
        this.sysMessageLogService.changeMessageStatus(messageId, SysMessageLogStatusEnum.SENDING_FAILURE);
    }

    private void handleSuccess(Long messageId) {
        this.sysMessageLogService.changeMessageStatus(messageId, SysMessageLogStatusEnum.SENDING_SUCCESS);
    }

    private void returnFailure(Long messageId) {
        check(messageId);
        this.sysMessageLogService.changeMessageStatus(messageId, SysMessageLogStatusEnum.HANDLING_FAILURE);
    }

    private Long newMessageLog(String title, SysRabbitMessage message) throws IOException {
        String stringMessage = ObjectUtils.serializeObjectToString(message);
        SysMessageLog sysMessageLog = this.sysMessageLogService.create();
        sysMessageLog.setTitle(title);
        sysMessageLog.setContent(stringMessage);
        return this.sysMessageLogService.add(sysMessageLog).getMessageId();
    }

    @Autowired
    public MessageSenderServiceImpl(RabbitTemplate template, SysMessageLogService sysMessageLogService) {
        this.template = template;
        this.sysMessageLogService = sysMessageLogService;
        this.template.setConfirmCallback(this.confirmCallback);
        this.template.setReturnCallback(this.returnCallback);
    }

    /**
     * 发送rabbit消息
     *
     * @param exchange        交换机
     * @param routingKey      路由键
     * @param message         消息
     */
    @Override
    public void sendMessage(String exchange, String routingKey, SysRabbitMessage message, Long messageId) {
        // 消息ID
        CorrelationData correlationData = new CorrelationData(String.valueOf(messageId));
        this.template.convertAndSend(exchange, routingKey, message, correlationData);
    }

    /**
     * 发送rabbit消息
     *
     * @param message   消息
     * @param messageId 消息ID
     */
    @Override
    public void sendMessage(SysRabbitMessage message, Long messageId) {
        this.sendMessage(this.defaultExchange, this.defaultRoutingKey, message, messageId);
    }

    /**
     * 发送rabbit消息
     *
     * @param message 消息
     */
    @Override
    public void sendMessage(String title, SysRabbitMessage message) {
        // 记录系统消息日志
        Long messageId;
        try {
            messageId = this.newMessageLog(title, message);
        } catch (IOException e) {
            log.error("创建消息日志出现异常：{}", e);
            throw new FrameworkException("创建消息日志出现异常");
        }
        this.sendMessage(this.defaultExchange, this.defaultRoutingKey, message, messageId);
    }

}
